package net.wicp.tams.common.others.kafka;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import net.wicp.tams.common.Conf;
import net.wicp.tams.common.apiext.StringUtil;

public abstract class KafkaTools {
	private static Properties prop = null;// 幂等的，再做一次也无防,不做双重检查了

	public static Properties getProps(boolean isProducer) {
		if (prop == null) {
			synchronized (new Object()) {
				Properties temp = new Properties();
				Map<String, String> nameToType = new HashMap<>();
				nameToType.put("retries", "integer");
				nameToType.put("max.block.ms", "long");
				nameToType.put("max.in.flight.requests.per.connection", "integer");
				nameToType.put("linger.ms", "integer");
				nameToType.put("batch.size", "integer");
				nameToType.put("buffer.memory", "integer");
				nameToType.put("max.request.size", "integer");
				nameToType.put("receive.buffer.bytes", "integer");
				nameToType.put("request.timeout.ms", "integer");
				nameToType.put("send.buffer.bytes", "integer");
				nameToType.put("connections.max.idle.ms", "integer");
				
				Map<String, String> paramsMap = Conf.getPre("common.others.kafka."+(isProducer?"producer.":"consumer."), true);
				paramsMap.putAll(Conf.getPre("common.others.kafka.common.", true));
				for (String key : paramsMap.keySet()) {
					if (StringUtil.isNotNull(paramsMap.get(key))) {
						if (nameToType.containsKey(key)) {
							if ("integer".equals(nameToType.get(key))) {
								temp.put(key, Integer.parseInt(paramsMap.get(key)));
							} else if ("long".equals(nameToType.get(key))) {
								temp.put(key, Long.parseLong(paramsMap.get(key).replace("l", "").replace("L", "")));
							}
						} else {
							temp.put(key, paramsMap.get(key));
						}
					}
				}
				prop = temp;
			}
		}
		return prop;
	}

	public static String getValueProp(Class<?> classz, boolean isProducer) {
		if ("java.lang.String".equals(classz.getName())) {
			return isProducer ? "org.apache.kafka.common.serialization.StringSerializer"
					: "org.apache.kafka.common.serialization.StringDeserializer";
		} else if ("[B".equals(classz.getName())) {
			return isProducer ? "org.apache.kafka.common.serialization.ByteArraySerializer"
					: "org.apache.kafka.common.serialization.ByteArrayDeserializer";
		} else {
			throw new IllegalArgumentException("不支持的类型");
		}
	}
}
